package net.butfly.bus.filter;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.FutureTask;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import net.butfly.albacore.exception.SystemException;
import net.butfly.bus.Constants;
import net.butfly.bus.Request;
import net.butfly.bus.Response;
import net.butfly.bus.context.Context;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ThreadControlFilter extends FilterBase implements Filter {
	private static Logger logger = LoggerFactory.getLogger(ThreadControlFilter.class);
	private ExecutorService executor;
	private long timeout;

	@Override
	public void initialize(Map<String, String> params) {
		super.initialize(params);
		String val = params.get("corePoolSize");
		int corePoolSize = null == val ? Constants.Async.DEFAULT_CORE_POOL_SIZE : Integer.parseInt(val);
		val = params.get("maxPoolSize");
		int maxPoolSize = null == val ? Constants.Async.DEFAULT_MAX_POOL_SIZE : Integer.parseInt(val);
		val = params.get("keepAliveTime");
		long keepAliveTime = null == val ? Constants.Async.DEFAULT_ALIVE_TIME : Long.parseLong(val);
		this.executor = new ThreadPoolExecutor(corePoolSize, maxPoolSize, keepAliveTime, TimeUnit.MILLISECONDS,
				new SynchronousQueue<Runnable>());
		val = params.get("timeout");
		this.timeout = null == val ? Constants.Async.DEFAULT_TIMEOUT : Long.parseLong(val);
	}

	@Override
	public void execute(Request request, Response response) throws Exception {
		FutureTask<AsyncResult> task = new FutureTask<AsyncResult>(new AsyncTask(request, response, Context.CURRENT));
		try {
			executor.execute(task);
		} catch (RejectedExecutionException e) {
			logger.warn("async task executing rejected for pool saturated...");
			throw new SystemException(Constants.SystemError.SATURATED, "请求数超过了上限。");
		}
		try {
			AsyncResult r = null;
			if (timeout > 0)
				r = task.get(timeout, TimeUnit.MILLISECONDS);
			else
				r = task.get();
			logger.debug("请求控制：请求成功完成");
			response.copyFrom(r.response);
			Context.CURRENT.clear();
			Context.CURRENT.putAll(r.context);
		} catch (InterruptedException e) {
			task.cancel(true);
			throw new SystemException(Constants.SystemError.INTERRUPTED, e);
		} catch (TimeoutException e) {
			task.cancel(true);
			throw new SystemException(Constants.SystemError.TIMEOUT, "请求超时。");
		} catch (ExecutionException e) {
			task.cancel(true);

			Throwable ex = e.getCause();
			if (ex instanceof Exception)
				throw (Exception) ex;
			logger.error("Unhandlable exception: ", ex);
			throw new SystemException(Constants.SystemError.UNKNOW_CAUSE, ex);
		}
	}

	private final class AsyncTask implements Callable<AsyncResult> {
		final Request request;
		final Response response;
		final Map<String, Serializable> context = new HashMap<String, Serializable>();

		public AsyncTask(Request request, Response response, Context context) {
			super();
			request.context(Context.serialize(Context.toMap()));
			this.request = request;
			this.response = response;
			this.context.putAll(context);
		}

		public AsyncResult call() throws Exception {
			Context.CURRENT.putAll(this.context);
			AsyncResult r = new AsyncResult();
			ThreadControlFilter.super.execute(request, response);
			r.response = response;
			r.context.putAll(Context.serialize(Context.toMap()));
			return r;
		}
	}

	private final class AsyncResult {
		Response response;
		Map<String, Serializable> context = new HashMap<String, Serializable>();
	}
}
